package com.bclz.parallelism

import java.util.concurrent.{Callable, CountDownLatch, ExecutorService}
import java.util.concurrent.atomic.AtomicReference

/**
  * 再无错误的前提下，提供一个非阻塞的Par
  * 当出现异常，latch.await()一直阻塞下去
  */
object Nonblocking{

  sealed trait Future[+A]{
    //只能在parallelism包下访问
    //f可以当初一个结果收集器
    private[parallelism] def apply(f:A=>Unit):Unit
  }
  type Par[+A]= ExecutorService=>Future[A]


  def run[A](e:ExecutorService)(p:Par[A]):A={

    val result=new AtomicReference[A]()
    val latch=new CountDownLatch(1)


    p(e)(a=>{
        result.set(a)
        latch.countDown
    })
    //这个位置可能会阻塞，但只需等待结果set到reference就释放了
    latch.await()
    result.get
  }

  def unit[A](a:A):Par[A]=
    _=>new Future[A] {
      override  def apply(f: A => Unit): Unit = f(a)
    }

  //延迟加载
  def fork[A](p: =>Par[A]):Par[A]=
    es=>new Future[A] {
      override  def apply(f: A => Unit): Unit =
        asynEval(es)(p(es)(f))
    }

  def map2[A,B,C](p1:Par[A],p2:Par[B])(c:(A,B)=>C):Par[C]=
    es=>new Future[C] {
      override  def apply(f: C => Unit): Unit ={

        var optionalA:Option[A]=None
        var optionalB:Option[B]=None
        val actor=Actor[Either[A,B]](es) {

          case Left(a)=>   //A结果先到，则等待B结果；A的结果后到，调用c获取结果C给回调f
            optionalB match {
              case None =>optionalA=Some(a)
              case Some(b) => asynEval(es)(f(c(a,b)))
            }
          case Right(b)=>  //同理A
            optionalA match {
              case None =>optionalB=Some(b)
              case Some(a) => asynEval(es)(f(c(a,b)))
            }

        }
        p1(es)(a => actor ! Left(a))
        p2(es)(b => actor ! Right(b))

      }
    }

  def parMap[A,B](list: List[A])(f:A=>B):Par[List[B]]={
    val listPar = list.map(asyncF(f))
    sequence(listPar)
  }

  def asyncF[A, B](f: A => B): A => Par[B] =
    a => lazyUnit(f(a))

  def lazyUnit[A](a: =>A):Par[A]=
    fork(unit(a))

  //容易栈溢出
//  def sequence[A](list:List[Par[A]]):Par[List[A]]=
//    list match {
//      case Nil =>unit(Nil)
//      case h::t=>map2(h,fork(sequence(t)))(_::_)
//    }

  def sequenceBalanced[A](as: IndexedSeq[Par[A]]): Par[IndexedSeq[A]] = fork {
    if (as.isEmpty) unit(Vector())
    else if (as.length == 1) map(as.head)(a => Vector(a))
    else {
      val (l,r) = as.splitAt(as.length/2)
      map2(sequenceBalanced(l), sequenceBalanced(r))(_ ++ _)
    }
  }

  def map[A, B](p: Par[A])(f: A => B): Par[B] =
    es => new Future[B] {
      def apply(cb: B => Unit): Unit =
        p(es)(a => asynEval(es) {
          cb(f(a))
        })
    }

  def sequence[A](as: List[Par[A]]): Par[List[A]] =
    map(sequenceBalanced(as.toIndexedSeq))(_.toList)


  def asynEval(es:ExecutorService)(c: => Unit): Unit ={

    es.submit(new Callable[Unit] {
      override def call(): Unit = c
    })

  }

  def choice[A](cond:Par[Boolean])(p1:Par[A],p2:Par[A]):Par[A]=
    es=>if(run(es)(cond)) p1(es) else p2(es)

  def choiceN[A](choiceIndex:Par[Int])(parList:List[Par[A]]):Par[A]=
    es=>parList(run(es)(choiceIndex))(es)

  def choiceMap[K,V](choiceKey:Par[K])(parMap:Map[K,Par[V]]):Par[V]=
    es=>parMap(run(es)(choiceKey))(es)


  def chooser[A,B](choiceKey:Par[A])(f:A=>Par[B]):Par[B]=
    es=>f(run(es)(choiceKey))(es)

  def choiceViaChooser[A](cond:Par[Boolean])(p1:Par[A],p2:Par[A]):Par[A]=
    chooser(cond)(a=>if(a) p1 else p2 )

  def choiceNViaChooser[A](choiceIndex:Par[Int])(parList:List[Par[A]]):Par[A]=
    chooser(choiceIndex)(a=>parList(a))

//  def flatMap[A,B](a:Par[A])(f:A=>Par[B]):Par[B]=
//    es=>f(run(es)(a))(es)


}
